#include <ctime>
#include <cstdlib>
#include <iostream>
#include "MDS.h"
#include "debug.h"
//using namespace std;

int MDS::parse_conf()
{
	debug(SECTION_MDS, 10, "---> parse_conf\n");
	int ret = 0;
	FILE *fp;
	char opt_name[100];
	char opt_val[100];
	if ((fp = fopen(conf.c_str(), "r")) != NULL) {
		while (fscanf(fp, "%s\t=\t%s", opt_name, opt_val) != EOF) {
			debug(SECTION_MDS, 10, "CONFIG : %s = %s\n", opt_name, opt_val);
			if (!strcmp(opt_name, "master_id2uri")) {
				char id_[5];
				char uri_[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id_, uri_);
				master_uri = strdup(uri_);
				uri_to_session_id[master_uri] = 0;
			} else if (!strcmp(opt_name, "cs_id2uri")) {
				char id_[5];
				char uri_[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id_, uri_);
				uint64_t cs_id = atoi(id_);
				cs_uri[cs_id] = strdup(uri_);
				uri_to_session_id[cs_uri[cs_id]] = 0;
			} else if (!strcmp(opt_name, "mds_id2uri")) {
				char id_[5];
				char uri_[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id_, uri_);
				id = atoi(id_);
				uri = strdup(uri_);
			} else if (!strcmp(opt_name, "dir_meta_db_type")) {
				dir_meta_db_type = strdup(opt_val);
			} else if (!strcmp(opt_name, "dir_meta_db_path")) {
				dir_meta_db_path = strdup(opt_val);
			} else if (!strcmp(opt_name, "dir_data_db_type")) {
				dir_data_db_type = strdup(opt_val);
			} else if (!strcmp(opt_name, "dir_data_db_path")) {
				dir_data_db_path = strdup(opt_val);
			} else if (!strcmp(opt_name, "file_meta_db_type")) {
				file_meta_db_type = strdup(opt_val);
			} else if (!strcmp(opt_name, "file_meta_db_path")) {
				file_meta_db_path = strdup(opt_val);
			} else if (!strcmp(opt_name, "file_data_db_type")) {
				file_data_db_type = strdup(opt_val);
			} else if (!strcmp(opt_name, "file_data_db_path")) {
				file_data_db_path = strdup(opt_val);
			} else if (!strcmp(opt_name, "worker_pool_size")) {
				pool_size = atoi(opt_val);
			} else {
			}
		}
		fclose(fp);
	} else {
		ret = -1;
		debug(SECTION_MDS, 10, "Failed to open conf file : %s", conf.c_str());
		return ret;
	}
	return ret;
}

void MDS::prepare()
{
	debug(SECTION_MDS, 10, "---> prepare\n");
	//------1. kvdb init-----------
	dir_meta_db = KeyValueDB::create(dir_meta_db_type, dir_meta_db_path);
	dir_data_db = KeyValueDB::create(dir_data_db_type, dir_data_db_path);
	file_meta_db = KeyValueDB::create(file_meta_db_type, file_meta_db_path);
	file_data_db = KeyValueDB::create(file_data_db_type, file_data_db_path);
	dir_meta_db->open();
	dir_data_db->open();
	file_meta_db->open();
	file_data_db->open();

	string nodeidstr("NODEID");
	string nodeidval;
	nodeid_lock.lock();
	if ( 1 == dir_meta_db->get(nodeidstr, &nodeidval)) {
		nodeidval = to_string(nodeid);
		dir_meta_db->set(nodeidstr, &nodeidval);
	} else {
		nodeid = stoull(nodeidval);
	}
	nodeid_lock.unlock();
	// create root dir
	debug(SECTION_MDS, 10, " check and create root dir...\n");
	string root("/");
	string rootstr;
	if ( 1 == dir_meta_db->get(root, &rootstr)) {
		debug(SECTION_MDS, 10, " root dir is not exist, create it...\n");
		struct Dir dir;
		nodeid_lock.lock();
		dir.id = nodeid++;
		string nodeidstr("NODEID");
		string nodeidval = to_string(nodeid);
		dir_meta_db->set(nodeidstr, &nodeidval);
		nodeid_lock.unlock();

		strcpy(dir.name, "/");
		dir.ctime = clock();
		dir.mode = 777;	//TODO
		dir.uid = 0;
		dir.gid = 0;
		dir.encode(&rootstr);
		dir_meta_db->set(root, &rootstr);
		struct Dentry dentry1, dentry2;
		string denstr, denstr1, denstr2;
		dentry1.id = dir.id;
		strcpy(dentry1.name, ".");
		dentry1.type = 1;
		dentry1.encode(&denstr1);
		denstr += denstr1;
		denstr += '\n';
		dentry2.id = dir.id;
		strcpy(dentry2.name, "..");
		dentry2.type = 1;
		dentry2.encode(&denstr2);
		denstr += denstr2;
		denstr += '\n';
		string idstr = to_string(dir.id);
		dir_data_db->set(idstr, &denstr);
	} else {
		debug(SECTION_MDS, 10, " root : %s\n", rootstr.c_str());
	}
	//------2. transaction init----------

	//------3. lock service init--------

	//------4. rpc server init------
	debug(SECTION_MDS, 10, "init server rpc\n");
	uint32_t mem0_size = 1 << 30;
	mem0 = new char[mem0_size];
	mm0 = new MemoryManager(mem0, mem0_size, 256 << 10);
	mm0->Init();
	rpc_server = new RdmaRpc(*mm0);
	rpc_server->Init(RPC_SERVER | RPC_RDMA | RPC_SHARED_RECV_QUEUE);
	rpc_server->BindSess(uri.c_str());
	debug(SECTION_MDS, 10, "register file_op_service\n");
	rpc_server->RpcServiceRegister(MDS_FILE_OP_SERVICE, file_op_service);
	rpc_server->RpcServiceProviderRegister(MDS_FILE_OP_SERVICE, this);
	//pthread_create(&(rpc_server_th), 0, rpc_server_main, this);

	//------5. rpc client init------
	debug(SECTION_MDS, 10, "init client rpc\n");
	uint32_t mem1_size = 1 << 30;
	mem1 = new char[mem1_size];
	mm1 = new MemoryManager(mem1, mem1_size, 256 << 10);
	mm1->Init();
	rpc_client = new RdmaRpc(*mm1);
	rpc_client->Init(RPC_RDMA | RPC_SHARED_RECV_QUEUE);
	uri_to_session_id[master_uri] = rpc_client->ConnectSess(master_uri.c_str(), NULL, NULL);
	for (int i = 0; i < cs_uri.size(); i++) {
		uri_to_session_id[cs_uri[i]] = rpc_client->ConnectSess(cs_uri[i].c_str(), NULL, NULL);
	}
	//pthread_create(&(rpc_client_th), 0, rpc_client_main, this);

	//------6. worker thread pool init------
	worker_thread_pool.resize(pool_size);
/*
	for (int i = 0; i < pool_size; i++) {
		pthread_create(&(worker_thread_pool[i]), 0, worker_process, this);
	}
*/
	//-----------------------------
	state = ACTIVE;
	debug(SECTION_MDS, 10, "<------------prepare\n");
}
void MDS::run()
{
	pthread_create(&(rpc_server_th), 0, rpc_server_main, this);
	pthread_create(&(rpc_client_th), 0, rpc_client_main, this);
	for (int i = 0; i < pool_size; i++) {
		pthread_create(&(worker_thread_pool[i]), 0, worker_process, this);
	}
}

void MDS::exit()
{
	pthread_join(rpc_server_th, NULL);
	pthread_join(rpc_client_th, NULL);
	for (int i = 0; i < pool_size; i++) {
		pthread_join(worker_thread_pool[i], NULL);
	}
}

void *MDS::rpc_server_main(void *arg)
{
	debug(SECTION_MDS, 10, "------------>rpc_server_main\n");
	pthread_setname_np(pthread_self(), "rpc_server_th");
	MDS *mds = static_cast<MDS *>(arg);
	clock_t now, last_active;
	now = clock();
	last_active = now;
	struct RpcExecution *ex = NULL;
	while (mds->state == ACTIVE) {
		now = clock();
		if (((float)(now - last_active)/CLOCKS_PER_SEC) > 1.0) {
			mds->rpc_server->AcceptSess();
			last_active = now;
		}
		mds->rpc_server->RpcServerHandleRx();
		mds->rpc_server_tx_q_lock.lock();
		if (!mds->rpc_server_tx_q.empty()) {
			debug(SECTION_MDS, 10, "dequeue rpc_server_tx_q\n");
			ex = mds->rpc_server_tx_q.front();
			mds->rpc_server_tx_q.pop_front();
			mds->rpc_server_tx_q_lock.unlock();
			mds->rpc_server->RpcServiceComplete(ex);
		} else {
			mds->rpc_server_tx_q_lock.unlock();
		}
		mds->rpc_server->RpcServerHandleTx();
	}
}

void MDS::file_op_service(void* self_, void* ex_, void* mds_)
{
	debug(SECTION_MDS, 10, "------------>file_op_service\n");
	RpcExecution *ex = static_cast<RpcExecution *>(ex_);
	MDS *mds = static_cast<MDS *>(mds_);
	debug(SECTION_MDS, 10, "enqueue rpc_server_rx_q\n");
	mds->rpc_server_rx_q_lock.lock();
	mds->rpc_server_rx_q.push_back(ex);
	mds->rpc_server_rx_q_lock.unlock();
}

void *MDS::rpc_client_main(void *arg)
{
	debug(SECTION_MDS, 10, "------------>rpc_client_main\n");
	pthread_setname_np(pthread_self(), "rpc_client_th");
	MDS *mds = static_cast<MDS *>(arg);
	struct rpc_msg_info *msg_info = NULL;
	while (mds->state == ACTIVE) {
		mds->rpc_client_tx_q_lock.lock();
		if (!mds->rpc_client_tx_q.empty()) {
			debug(SECTION_MDS, 10, "dequeue rpc_client_tx_q\n");
			msg_info = mds->rpc_client_tx_q.front();
			mds->rpc_client_tx_q.pop_front();
			mds->rpc_client_tx_q_lock.unlock();
			RpcMsg *rpc_header = static_cast<RpcMsg *>(mds->rpc_client->AllocMsg());
			mds->rpc_client->PrepareMsg(rpc_header, msg_info->rpc_type);
			struct dss_msg *dss_header = reinterpret_cast<struct dss_msg*>(rpc_header->data);
			dss_header->client_id = msg_info->client_id;
			dss_header->msg_type = msg_info->msg_type;
			dss_header->msg_id = msg_info->msg_id;
			dss_header->data_len = msg_info->data_len;
			if (msg_info->data_len) {
				memcpy(dss_header->data, msg_info->data, msg_info->data_len);
			}
			rpc_header->data_len = sizeof(struct dss_msg) + dss_header->data_len;
			mds->rpc_client->RpcCliWriteReq(mds->uri_to_session_id[msg_info->uri], rpc_header, msg_info->cb, msg_info);
		} else {
			mds->rpc_client_tx_q_lock.unlock();
		}
		mds->rpc_client->RpcCliHandleRx();
	}
}

void *MDS::worker_process(void *arg)
{
	uint64_t pthread_id = pthread_self();
	debug(SECTION_MDS, 10, "------------>worker_process--%lld\n", pthread_id);
	pthread_setname_np(pthread_id, "worker_th");
	MDS *mds = static_cast<MDS *>(arg);
	struct RpcExecution *ex = NULL;
	while (1) {
		mds->rpc_server_rx_q_lock.lock();
		if (!mds->rpc_server_rx_q.empty()) {
			debug(SECTION_MDS, 10, "<worker--%lld> dequeue rpc_server_rx_q\n", pthread_id);
			ex = mds->rpc_server_rx_q.front();
			mds->rpc_server_rx_q.pop_front();
			mds->rpc_server_rx_q_lock.unlock();
			RpcMsg *rpc_req = ex->req_msg;
			RpcMsg *rpc_resp = ex->resp_msg;
			struct dss_msg *msg = reinterpret_cast<struct dss_msg*>(rpc_req->data);
			struct dss_msg *msg_resp = reinterpret_cast<struct dss_msg*>(rpc_resp->data);
			switch (msg->msg_type)
			{
				case MDS_FILE_MKNOD:
				{
					debug(SECTION_MDS, 10, "MKNOD ...\n");
					struct FileOpMKNOD *op = (struct FileOpMKNOD *)msg->data;
					mds->handle_mknod(op, msg_resp);
					break;
				}
				case MDS_FILE_MKDIR:
				{
					debug(SECTION_MDS, 10, "MKDIR ...\n");
					struct FileOpMKDIR *op = (struct FileOpMKDIR *)msg->data;
					mds->handle_mkdir(op, msg_resp);
					break;
				}
				case MDS_FILE_UNLINK:
				{
					debug(SECTION_MDS, 10, "UNLINK ...\n");
					struct FileOpUNLINK *op = (struct FileOpUNLINK *)msg->data;
					mds->handle_unlink(op, msg_resp);
					break;
				}
				case MDS_FILE_RMDIR:
				{
					debug(SECTION_MDS, 10, "RMDIR ...\n");
					struct FileOpRMDIR *op = (struct FileOpRMDIR *)msg->data;
					mds->handle_rmdir(op, msg_resp);
					break;
				}
				case MDS_FILE_RENAME:
				{
					debug(SECTION_MDS, 10, "RENAME ...\n");
					struct FileOpRENAME *op = (struct FileOpRENAME *)msg->data;
					mds->handle_rename(op, msg_resp);
					break;
				}
				case MDS_FILE_STAT:
				{
					debug(SECTION_MDS, 10, "STAT ...\n");
					struct FileOpSTAT *op = (struct FileOpSTAT *)msg->data;
					mds->handle_stat(op, msg_resp);
					break;
				}
				case MDS_FILE_CHMOD:
				{
					debug(SECTION_MDS, 10, "CHMOD ...\n");
					struct FileOpCHMOD *op = (struct FileOpCHMOD *)msg->data;
					mds->handle_chmod(op, msg_resp);
					break;
				}
				case MDS_FILE_CHOWN:
				{
					debug(SECTION_MDS, 10, "CHOWN ...\n");
					struct FileOpCHOWN *op = (struct FileOpCHOWN *)msg->data;
					mds->handle_chown(op, msg_resp);
					break;
				}
				case MDS_FILE_OPEN:
				{
					debug(SECTION_MDS, 10, "OPEN ...\n");
					struct FileOpOPEN *op = (struct FileOpOPEN *)msg->data;
					mds->handle_open(op, msg_resp);
					break;
				}
				case MDS_FILE_READ:
				{
					debug(SECTION_MDS, 10, "READ ...\n");
					struct FileOpREAD *op = (struct FileOpREAD *)msg->data;
					mds->handle_read(op, msg_resp);
					break;
				}
				case MDS_FILE_WRITE:
				{
					debug(SECTION_MDS, 10, "WRITE ...\n");
					struct FileOpWRITE *op = (struct FileOpWRITE *)msg->data;
					mds->handle_write(op, msg_resp);
					break;
				}
				case MDS_FILE_TRUNCATE:
				{
					debug(SECTION_MDS, 10, "TRUNCATE ...\n");
					struct FileOpTRUNCATE *op = (struct FileOpTRUNCATE *)msg->data;
					mds->handle_truncate(op, msg_resp);
					break;
				}
				case MDS_FILE_CLOSE:
				{
					debug(SECTION_MDS, 10, "CLOSE ...\n");
					struct FileOpCLOSE *op = (struct FileOpCLOSE *)msg->data;
					mds->handle_close(op, msg_resp);
					break;
				}
				case MDS_FILE_FSYNC:
				{
					debug(SECTION_MDS, 10, "FSYNC ...\n");
					struct FileOpFSYNC *op = (struct FileOpFSYNC *)msg->data;
					mds->handle_fsync(op, msg_resp);
					break;
				}
				case MDS_FILE_READDIR:
				{
					debug(SECTION_MDS, 10, "READDIR ...\n");
					struct FileOpREADDIR *op = (struct FileOpREADDIR *)msg->data;
					mds->handle_readdir(op, msg_resp);
					break;
				}
				case MDS_FILE_STATFS:
				{
					debug(SECTION_MDS, 10, "STATFS ...\n");
					struct FileOpSTATFS *op = (struct FileOpSTATFS *)msg->data;
					mds->handle_statfs(op, msg_resp);
					break;
				}
				default:
				{
					debug(SECTION_MDS, 10, "unknown file op type !!!\n");
					break;
				}
			}
			msg_resp->msg_id = msg->msg_id;
			rpc_resp->status_code = SUCCESS;
			rpc_resp->data_len = sizeof(struct dss_msg) + msg_resp->data_len;
			debug(SECTION_MDS, 10, "enqueue rpc_server_tx_q\n");
			mds->rpc_server_tx_q_lock.lock();
			mds->rpc_server_tx_q.push_back(ex);
			mds->rpc_server_tx_q_lock.unlock();
		} else {
			mds->rpc_server_rx_q_lock.unlock();
		}
	}
}

void MDS::correct(const char *old_path, char *new_path)
{
	debug(SECTION_MDS, 10, "---> correct\n");
  int i, count = 0;
  int len = strlen(old_path);
  for (i = 0; i < len; i++) {
    if ((old_path[i] == '/') && (old_path[i + 1] == '/'))
      continue;
    new_path[count++] = old_path[i];
  }
  if (new_path[count - 1] == '/') {
    new_path[count - 1] = '\0';
  } else {
    new_path[count] = '\0';
  }
  if(new_path[0] == '\0')
  {
    new_path[0] = '/';
    new_path[1] = '\0';
  }
	debug(SECTION_MDS, 10, "old : %s\n", old_path);
	debug(SECTION_MDS, 10, "new : %s\n", new_path);
	debug(SECTION_MDS, 10, "<--- correct\n");
}

bool MDS::get_ppath_name(const char *path, char *parent, char *name)
{
	int i;
	debug(SECTION_MDS, 10, "---> get parent dir\n");
  if ((path == NULL) || (parent == NULL)) {
		debug(SECTION_MDS, 10, "path or parent is NULL!!\n");
		debug(SECTION_MDS, 10, "<--- get parent dir\n");
    return false;
  } else {
    strcpy(parent, path);
    uint64_t lengthPath = strlen(path);
    if ((lengthPath == 1) && (path[0] == '/')) {
			debug(SECTION_MDS, 10, "path is root dir\n");
			parent[0] = '/';
			parent[1] = '\0';
			name[0] = '/';
			name[1] = '\0';
			debug(SECTION_MDS, 10, "<--- get parent dir\n");
      return false;
    } else {
      bool resultCut = false;
      for (i = lengthPath - 1; i >= 0; i--) {
        if (path[i] == '/') {
          parent[i] = '\0';
          resultCut = true;
          break;
        }
      }
      if (resultCut == false) {
				debug(SECTION_MDS, 10, "error path : %s\n", path);
				debug(SECTION_MDS, 10, "<--- get parent dir\n");
        return false;
      } else {
				for (int j = i+1; j <= lengthPath; j++) {
					name[j-i-1] = path[j];
				}
        if (parent[0] == '\0') {
          parent[0] = '/';
          parent[1] = '\0';
					debug(SECTION_MDS, 10, "path : %s , parent : %s .\n", path, parent);
					debug(SECTION_MDS, 10, "<--- get parent dir\n");
          return true;
        } else {
					debug(SECTION_MDS, 10, "path : %s , parent : %s .\n", path, parent);
					debug(SECTION_MDS, 10, "<--- get parent dir\n");
          return true;
        }
      }
    }
  }
}


/*************************************/

void MDS::handle_mknod(struct FileOpMKNOD *op, struct dss_msg *m_resp)
{
	//------ prepare -----------
	char path[MAX_PATH_LENGTH], ppath[MAX_PATH_LENGTH], name[MAX_NAME_LENGTH];
	correct(op->path, path);
	if (0 == strcmp(path, "/")) {
		m_resp->status_code = MDS_EEXIST;
		m_resp->msg_type = MDS_FILE_MKNOD_REPLY;
		m_resp->data_len = 0;
		return;
	}
	get_ppath_name(path, ppath, name);
	string pathstr(path);
	string ppathstr(ppath);

	struct Inode inode;
	struct Dir pdir;
	string inodestr, dirstr, pdirstr;
	string pdirdatastr;

	uint64_t *path_lock_addr, *ppath_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	ppath_lock_addr = locks.PathToAddress(ppathstr);

	locks.WriteLock(path_lock_addr);
	locks.WriteLock(ppath_lock_addr);
	if ( 0 == dir_meta_db->get(pathstr, &dirstr)) {
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		m_resp->status_code = MDS_EEXIST;
		m_resp->msg_type = MDS_FILE_MKNOD_REPLY;
		m_resp->data_len = 0;
		return;
	} else if (1 == dir_meta_db->get(ppathstr, &pdirstr)) {
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_MKNOD_REPLY;
		m_resp->data_len = 0;
		return;
	}
	pdir.decode(pdirstr);
	string pidstr = to_string(pdir.id);
	dir_data_db->get(pidstr, &pdirdatastr);
	istringstream iss(pdirdatastr);
	string token;
	while(getline(iss, token, '\n')) {
		struct Dentry d;
		d.decode(token);
		if (strcmp(name, d.name) == 0) {
			locks.WriteUnlock(ppath_lock_addr);
			locks.WriteUnlock(path_lock_addr);
			m_resp->status_code = MDS_EEXIST;
			m_resp->msg_type = MDS_FILE_MKNOD_REPLY;
			m_resp->data_len = 0;
			return;
		}
	}

	uint64_t tx = transaction.TransBegin();
	transaction.TransLogContext(tx, NULL, 0);

	//---kv op---set(path)
	nodeid_lock.lock();
	inode.id = nodeid++;
	string nodeidstr("NODEID");
	string nodeidval = to_string(nodeid);
	dir_meta_db->set(nodeidstr, &nodeidval);
	nodeid_lock.unlock();
	strcpy(inode.name, name);
	inode.ctime = clock();
	inode.mtime = inode.ctime;
	inode.atime = inode.ctime;
	inode.mode = op->mode;
	inode.uid = 0;
	inode.gid = 0;
	inode.size = 0;
	inode.bsize = 4<<20;	//4M TODO
	inode.encode(&inodestr);
	string filekey = pidstr + "_" + name;
	debug(SECTION_MDS, 10, "filekey = %s, inodestr = %s.\n", filekey.c_str(), inodestr.c_str());
	file_meta_db->set(filekey, &inodestr);

	//---kv op---update ppath's dentry list
	struct Dentry dentry;
	string dentrystr;
	dentry.id = inode.id;
	strcpy(dentry.name, inode.name);
	dentry.type = NORMAL_FILE;
	dentry.encode(&dentrystr);
	pdirdatastr += dentrystr;
	pdirdatastr += '\n';
	dir_data_db->set(pidstr, &pdirdatastr);
	//------------
	transaction.TransCommit(tx);
	locks.WriteUnlock(ppath_lock_addr);
	locks.WriteUnlock(path_lock_addr);
	m_resp->status_code = MDS_SUCCESS;
	m_resp->msg_type = MDS_FILE_MKNOD_REPLY;
	m_resp->data_len = 0;

}

void MDS::handle_mkdir(struct FileOpMKDIR *op, struct dss_msg *m_resp)
{
	//------ prepare -----------
	char path[MAX_PATH_LENGTH], ppath[MAX_PATH_LENGTH], name[MAX_NAME_LENGTH];
	correct(op->path, path);
	if (0 == strcmp(path, "/")) {
		m_resp->status_code = MDS_EEXIST;
		m_resp->msg_type = MDS_FILE_MKDIR_REPLY;
		m_resp->data_len = 0;
		return;
	}
	get_ppath_name(path, ppath, name);
	string pathstr(path);
	string ppathstr(ppath);

	struct Dir dir, pdir;
	string dirstr, pdirstr;
	string pdirdatastr;

	uint64_t *path_lock_addr, *ppath_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	ppath_lock_addr = locks.PathToAddress(ppathstr);

	locks.WriteLock(path_lock_addr);
	locks.WriteLock(ppath_lock_addr);
	if ( 0 == dir_meta_db->get(pathstr, &dirstr)) {
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		m_resp->status_code = MDS_EEXIST;
		m_resp->msg_type = MDS_FILE_MKDIR_REPLY;
		m_resp->data_len = 0;
		return;
	} else if (1 == dir_meta_db->get(ppathstr, &pdirstr)) {
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_MKDIR_REPLY;
		m_resp->data_len = 0;
		return;
	}
	pdir.decode(pdirstr);
	string pidstr = to_string(pdir.id);
	dir_data_db->get(pidstr, &pdirdatastr);
	istringstream iss(pdirdatastr);
	string token;
	while(getline(iss, token, '\n')) {
		struct Dentry d;
		d.decode(token);
		if (strcmp(name, d.name) == 0) {
			locks.WriteUnlock(ppath_lock_addr);
			locks.WriteUnlock(path_lock_addr);
			m_resp->status_code = MDS_EEXIST;
			m_resp->msg_type = MDS_FILE_MKDIR_REPLY;
			m_resp->data_len = 0;
			return;
		}
	}

	uint64_t tx = transaction.TransBegin();
	transaction.TransLogContext(tx, NULL, 0);

	//---kv op---set(path)
	nodeid_lock.lock();
	dir.id = nodeid++;
	string nodeidstr("NODEID");
	string nodeidval = to_string(nodeid);
	dir_meta_db->set(nodeidstr, &nodeidval);
	nodeid_lock.unlock();
	strcpy(dir.name, name);
	dir.ctime = clock();
	dir.mode = op->mode;
	dir.uid = 0;
	dir.gid = 0;
	dir.encode(&dirstr);
	dir_meta_db->set(pathstr, &dirstr);

	//---kv op---init path's dentry list
	struct Dentry dentry1, dentry2;
	string denstr, denstr1, denstr2;
	dentry1.id = dir.id;
	strcpy(dentry1.name, ".");
	dentry1.type = DIRECTORY;
	dentry1.encode(&denstr1);
	denstr += denstr1;
	denstr += '\n';
	dentry2.id = pdir.id;
	strcpy(dentry2.name, "..");
	dentry2.type = DIRECTORY;
	dentry2.encode(&denstr2);
	denstr += denstr2;
	denstr += '\n';
	string idstr = to_string(dir.id);
	dir_data_db->set(idstr, &denstr);

	//---kv op---update ppath's dentry list
	struct Dentry dentry;
	string dentrystr;
	dentry.id = dir.id;
	strcpy(dentry.name, dir.name);
	dentry.type = DIRECTORY;
	dentry.encode(&dentrystr);
	pdirdatastr += dentrystr;
	pdirdatastr += '\n';
	dir_data_db->set(pidstr, &pdirdatastr);
	//------------
	transaction.TransCommit(tx);
	locks.WriteUnlock(ppath_lock_addr);
	locks.WriteUnlock(path_lock_addr);
	m_resp->status_code = MDS_SUCCESS;
	m_resp->msg_type = MDS_FILE_MKDIR_REPLY;
	m_resp->data_len = 0;
}

void MDS::handle_unlink(struct FileOpUNLINK *op, struct dss_msg *m_resp)
{
	//------ prepare -----------
	char path[MAX_PATH_LENGTH], ppath[MAX_PATH_LENGTH], name[MAX_NAME_LENGTH];
	correct(op->path, path);
	get_ppath_name(path, ppath, name);
	if (0 == strcmp(path, "/")) {
		m_resp->status_code = MDS_EISDIR;
		m_resp->msg_type = MDS_FILE_UNLINK_REPLY;
		m_resp->data_len = 0;
		return;
	}
	string pathstr(path);
	string ppathstr(ppath);

	struct Dir pdir;
	string pdirstr;
	string pdirdatastr;

	uint64_t *path_lock_addr, *ppath_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	ppath_lock_addr = locks.PathToAddress(ppathstr);

	locks.WriteLock(path_lock_addr);
	locks.WriteLock(ppath_lock_addr);
	if ( 1 == dir_meta_db->get(ppathstr, &pdirstr)) {
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_UNLINK_REPLY;
		m_resp->data_len = 0;
		return;
	}
	pdir.decode(pdirstr);
	string pidstr = to_string(pdir.id);
	dir_data_db->get(pidstr, &pdirdatastr);
	istringstream iss(pdirdatastr);
	string token;
	struct Dentry dentry;
	int flag = 0;
	string pdirdatastr1;
	while(getline(iss, token, '\n')) {
		dentry.decode(token);
		if (strcmp(dentry.name, name) == 0) {
			if (dentry.type == NORMAL_FILE) {
				flag = 1;
				debug(SECTION_MDS, 10, "flag = %d\n", flag);
				//break;
			} else {
				locks.WriteUnlock(ppath_lock_addr);
				locks.WriteUnlock(path_lock_addr);
				m_resp->status_code = MDS_EISDIR;
				m_resp->msg_type = MDS_FILE_UNLINK_REPLY;
				m_resp->data_len = 0;
				return;
			}
		} else {
			pdirdatastr1 += token;
			pdirdatastr1 += "\n";
		}
	}
	if(0 == flag) {
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_UNLINK_REPLY;
		m_resp->data_len = 0;
		return;
	}
	uint64_t tx = transaction.TransBegin();
	transaction.TransLogContext(tx, NULL, 0);
	cinodes_lock.lock();
	if (cinodes.find(dentry.id) != cinodes.end()) {
		cinodes[dentry.id].unlink = true;
		cinodes_lock.unlock();
	} else {
		cinodes_lock.unlock();
		/** file_meta_db and file_data_db delete**/
		struct Inode inode;
		string filekey = pidstr + "_" + name;
		string inodestr;
		file_meta_db->get(filekey, &inodestr);
		inode.decode(inodestr);
		debug(SECTION_MDS, 10, "size = %d , bsize = %d\n", inode.size, inode.bsize);
		int64_t endblkno = inode.size / inode.bsize;
		if (inode.size % inode.bsize == 0)
			endblkno--;
		debug(SECTION_MDS, 10, "endblkno = %d\n", endblkno);
		for (int64_t i = 0; i <= endblkno; i++) {
			debug(SECTION_MDS, 10, "inodeid = %d , i = %d\n", inode.id, i);
			string fileblkkey = to_string(inode.id) + "_" + to_string(i);
			debug(SECTION_MDS, 10, "fileblkkey = %s\n", fileblkkey.c_str());
			string fblkstr;
			struct Fileblk fblk;
			int ret = file_data_db->get(fileblkkey, &fblkstr);
			debug(SECTION_MDS, 10, "ret = %d, fblkstr = %s\n", ret, fblkstr.c_str());
			fblk.decode(fblkstr);
			delete_chunk(&(fblk.chunkid));
			file_data_db->remove(fileblkkey);
		}
		file_meta_db->remove(filekey);
	}
	dir_data_db->set(pidstr, &pdirdatastr1);

	//------------
	transaction.TransCommit(tx);
	locks.WriteUnlock(ppath_lock_addr);
	locks.WriteUnlock(path_lock_addr);
	m_resp->status_code = MDS_SUCCESS;
	m_resp->msg_type = MDS_FILE_UNLINK_REPLY;
	m_resp->data_len = 0;

}

void MDS::handle_rmdir(struct FileOpRMDIR *op, struct dss_msg *m_resp)
{
	//------ prepare -----------
	char path[MAX_PATH_LENGTH], ppath[MAX_PATH_LENGTH], name[MAX_NAME_LENGTH];
	correct(op->path, path);
	get_ppath_name(path, ppath, name);
	if (0 == strcmp(path, "/")) {
		debug(SECTION_MDS, 10, "rmdir error, path is root dir\n");
		m_resp->status_code = MDS_EBUSY;
		m_resp->msg_type = MDS_FILE_RMDIR_REPLY;
		m_resp->data_len = 0;
		return;
	}

	string pathstr(path);
	string ppathstr(ppath);

	struct Dir dir, pdir;
	string dirstr, pdirstr;
	string dirdatastr, pdirdatastr;

	uint64_t *path_lock_addr, *ppath_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	ppath_lock_addr = locks.PathToAddress(ppathstr);

	locks.WriteLock(path_lock_addr);
	locks.WriteLock(ppath_lock_addr);
	if ( 1 == dir_meta_db->get(pathstr, &dirstr)) {
		debug(SECTION_MDS, 10, "rmdir error, path dir unexist.\n");
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_RMDIR_REPLY;
		m_resp->data_len = 0;
		return;
	}
	dir.decode(dirstr);
	string idstr = to_string(dir.id);
	dir_data_db->get(idstr, &dirdatastr);
	istringstream iss(dirdatastr);
	string token;
	struct Dentry dentry;
	while(getline(iss, token, '\n')) {
		dentry.decode(token);
		if (strcmp(dentry.name, ".") && strcmp(dentry.name, "..")) {
			debug(SECTION_MDS, 10, "rmdir error, path dir is not empty.\n");
			locks.WriteUnlock(ppath_lock_addr);
			locks.WriteUnlock(path_lock_addr);
			m_resp->status_code = MDS_ENOTEMPTY;
			m_resp->msg_type = MDS_FILE_RMDIR_REPLY;
			m_resp->data_len = 0;
			return;
		}
	}
	uint64_t tx = transaction.TransBegin();
	transaction.TransLogContext(tx, NULL, 0);
	dir_data_db->remove(idstr);
	dir_meta_db->remove(pathstr);
	dir_meta_db->get(ppathstr, &pdirstr);
	pdir.decode(pdirstr);
	string pidstr = to_string(pdir.id);
	dir_data_db->get(pidstr, &pdirdatastr);
	istringstream iss1(pdirdatastr);
	string pdirdatastr1;
	while(getline(iss1, token, '\n')) {
		dentry.decode(token);
		if (strcmp(dentry.name, name)) {
			pdirdatastr1 += token;
			pdirdatastr1 += "\n";
		}
	}
	dir_data_db->set(pidstr, &pdirdatastr1);

	//------------
	transaction.TransCommit(tx);
	debug(SECTION_MDS, 10, "rmdir success.\n");
	locks.WriteUnlock(ppath_lock_addr);
	locks.WriteUnlock(path_lock_addr);
	m_resp->status_code = MDS_SUCCESS;
	m_resp->msg_type = MDS_FILE_RMDIR_REPLY;
	m_resp->data_len = 0;

}


void MDS::handle_stat(struct FileOpSTAT *op, struct dss_msg *m_resp)
{
	//------ prepare -----------
	char path[MAX_PATH_LENGTH], ppath[MAX_PATH_LENGTH], name[MAX_NAME_LENGTH];
	correct(op->path, path);
	get_ppath_name(path, ppath, name);
	string pathstr(path);
	string ppathstr(ppath);
	string dirstr, pdirstr;
	uint64_t *path_lock_addr, *ppath_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	ppath_lock_addr = locks.PathToAddress(ppathstr);
	if (0 == strcmp(path, "/")) {
		locks.ReadLock(path_lock_addr);
		dir_meta_db->get(pathstr, &dirstr);
		struct FileOpSTATReply* stat_resp = reinterpret_cast<struct FileOpSTATReply*>(m_resp->data);
		stat_resp->type = DIRECTORY;
		stat_resp->dirinfo.decode(dirstr);
		m_resp->status_code = MDS_SUCCESS;
		m_resp->msg_type = MDS_FILE_STAT_REPLY;
		m_resp->data_len = sizeof(struct FileOpSTATReply);
		locks.ReadUnlock(path_lock_addr);
		return;
	}

	locks.ReadLock(path_lock_addr);
	locks.ReadLock(ppath_lock_addr);
	if ( 0 == dir_meta_db->get(pathstr, &dirstr)) {
		struct FileOpSTATReply* stat_resp = reinterpret_cast<struct FileOpSTATReply*>(m_resp->data);
		stat_resp->type = DIRECTORY;
		stat_resp->dirinfo.decode(dirstr);

		m_resp->status_code = MDS_SUCCESS;
		m_resp->msg_type = MDS_FILE_STAT_REPLY;
		m_resp->data_len = sizeof(struct FileOpSTATReply);
		locks.ReadUnlock(ppath_lock_addr);
		locks.ReadUnlock(path_lock_addr);
		return;
	} else if (1 == dir_meta_db->get(ppathstr, &pdirstr)) {
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_STAT_REPLY;
		m_resp->data_len = 0;
		locks.ReadUnlock(ppath_lock_addr);
		locks.ReadUnlock(path_lock_addr);
		return;
	}
	struct Dir pdir;
	string pdirdatastr;
	pdir.decode(pdirstr);
	string pidstr = to_string(pdir.id);
	dir_data_db->get(pidstr, &pdirdatastr);
	istringstream iss(pdirdatastr);
	string token;
	while(getline(iss, token, '\n')) {
		struct Dentry d;
		d.decode(token);
		if (strcmp(name, d.name) == 0) {
			string filestr = pidstr + "_" + name;
			string inodestr;
			file_meta_db->get(filestr, &inodestr);
			debug(SECTION_MDS, 10, "filekey = %s, inodestr = %s.\n", filestr.c_str(), inodestr.c_str());
			struct FileOpSTATReply* stat_resp = reinterpret_cast<struct FileOpSTATReply*>(m_resp->data);
			stat_resp->type = NORMAL_FILE;
			stat_resp->fileinfo.decode(inodestr);

			m_resp->status_code = MDS_SUCCESS;
			m_resp->msg_type = MDS_FILE_STAT_REPLY;
			m_resp->data_len = sizeof(struct FileOpSTATReply);
			locks.ReadUnlock(ppath_lock_addr);
			locks.ReadUnlock(path_lock_addr);
			return;
		}
	}
	m_resp->status_code = MDS_ENOENT;
	m_resp->msg_type = MDS_FILE_STAT_REPLY;
	m_resp->data_len = 0;
	locks.ReadUnlock(ppath_lock_addr);
	locks.ReadUnlock(path_lock_addr);
	return;
}

void MDS::handle_chmod(struct FileOpCHMOD *op, struct dss_msg *m_resp)
{

}

void MDS::handle_chown(struct FileOpCHOWN *op, struct dss_msg *m_resp)
{

}

void MDS::handle_open(struct FileOpOPEN *op, struct dss_msg *m_resp)
{
	//------ prepare -----------
	char path[MAX_PATH_LENGTH], ppath[MAX_PATH_LENGTH], name[MAX_NAME_LENGTH];
	correct(op->path, path);
	get_ppath_name(path, ppath, name);
	if (0 == strcmp(path, "/")) {
		m_resp->status_code = MDS_EISDIR;
		m_resp->msg_type = MDS_FILE_OPEN_REPLY;
		m_resp->data_len = 0;
		return;
	}
	string pathstr(path);
	string ppathstr(ppath);
	string dirstr, pdirstr;
	uint64_t *path_lock_addr, *ppath_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	ppath_lock_addr = locks.PathToAddress(ppathstr);

	locks.WriteLock(path_lock_addr);
	locks.WriteLock(ppath_lock_addr);
	if ( 0 == dir_meta_db->get(pathstr, &dirstr)) {
		m_resp->status_code = MDS_EISDIR;
		m_resp->msg_type = MDS_FILE_OPEN_REPLY;
		m_resp->data_len = 0;
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		return;
	} else if (1 == dir_meta_db->get(ppathstr, &pdirstr)) {
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_OPEN_REPLY;
		m_resp->data_len = 0;
		locks.WriteUnlock(ppath_lock_addr);
		locks.WriteUnlock(path_lock_addr);
		return;
	}
	struct Dir pdir;
	string pdirdatastr;
	pdir.decode(pdirstr);
	string pidstr = to_string(pdir.id);
	dir_data_db->get(pidstr, &pdirdatastr);
	istringstream iss(pdirdatastr);
	string token;
	while(getline(iss, token, '\n')) {
		struct Dentry d;
		d.decode(token);
		if (strcmp(name, d.name) == 0) {
			string filestr = pidstr + "_" + name;
			string inodestr;
			file_meta_db->get(filestr, &inodestr);
			debug(SECTION_MDS, 10, "filekey = %s, inodestr = %s.\n", filestr.c_str(), inodestr.c_str());
			struct FileOpOPENReply* open_resp = reinterpret_cast<struct FileOpOPENReply*>(m_resp->data);
			open_resp->fileinfo.decode(inodestr);
			cinodes_lock.lock();
			if (cinodes.find(open_resp->fileinfo.id) != cinodes.end()) {
				cinodes[open_resp->fileinfo.id].openctr++;
			} else {
				cinodes[open_resp->fileinfo.id].inode = open_resp->fileinfo;
				cinodes[open_resp->fileinfo.id].openctr = 1;
				strcpy(cinodes[open_resp->fileinfo.id].path, path);
				cinodes[open_resp->fileinfo.id].pdir_id = pdir.id;
				cinodes[open_resp->fileinfo.id].unlink = false;
			}
			cinodes_lock.unlock();
			m_resp->status_code = MDS_SUCCESS;
			m_resp->msg_type = MDS_FILE_OPEN_REPLY;
			m_resp->data_len = sizeof(struct FileOpOPENReply);
			locks.WriteUnlock(ppath_lock_addr);
			locks.WriteUnlock(path_lock_addr);
			return;
		}
	}
	if (op->flag & MDS_O_CREAT) {
		struct Inode inode;
		string inodestr;
		uint64_t tx = transaction.TransBegin();
		transaction.TransLogContext(tx, NULL, 0);

		//---kv op---set(path)
		nodeid_lock.lock();
		inode.id = nodeid++;
		string nodeidstr("NODEID");
		string nodeidval = to_string(nodeid);
		dir_meta_db->set(nodeidstr, &nodeidval);
		nodeid_lock.unlock();
		strcpy(inode.name, name);
		inode.ctime = clock();
		inode.mtime = inode.ctime;
		inode.atime = inode.ctime;
		inode.mode = op->mode;
		inode.uid = 0;
		inode.gid = 0;
		inode.size = 0;
		inode.bsize = 4<<20;	//4M TODO
		inode.encode(&inodestr);
		string filekey = pidstr + "_" + name;
		debug(SECTION_MDS, 10, "filekey = %s, inodestr = %s.\n", filekey.c_str(), inodestr.c_str());
		file_meta_db->set(filekey, &inodestr);

		//---kv op---update ppath's dentry list
		struct Dentry dentry;
		string dentrystr;
		dentry.id = inode.id;
		strcpy(dentry.name, inode.name);
		dentry.type = NORMAL_FILE;
		dentry.encode(&dentrystr);
		pdirdatastr += dentrystr;
		pdirdatastr += '\n';
		dir_data_db->set(pidstr, &pdirdatastr);
		//------------
		struct FileOpOPENReply* open_resp = reinterpret_cast<struct FileOpOPENReply*>(m_resp->data);
		open_resp->fileinfo = inode;

		cinodes_lock.lock();
		cinodes[inode.id].inode = inode;
		cinodes[inode.id].openctr = 1;
		strcpy(cinodes[inode.id].path, path);
		cinodes[inode.id].pdir_id = pdir.id;
		cinodes[inode.id].unlink = false;
		cinodes_lock.unlock();

		transaction.TransCommit(tx);
		m_resp->status_code = MDS_SUCCESS;
		m_resp->msg_type = MDS_FILE_OPEN_REPLY;
		m_resp->data_len = sizeof(struct FileOpOPENReply);
	} else {
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_OPEN_REPLY;
		m_resp->data_len = 0;
	}
	locks.WriteUnlock(ppath_lock_addr);
	locks.WriteUnlock(path_lock_addr);
	return;
}

void MDS::handle_read(struct FileOpREAD *op, struct dss_msg *m_resp)
{
	uint64_t	file_id = op->file_id;
	int64_t		offset = op->offset;
	int64_t		size = op->size;
	uint64_t	start_blk, end_blk;
	uint64_t	file_size, file_bsize;
	cinodes_lock.lock();
	if (cinodes.find(file_id) == cinodes.end()) {
		cinodes_lock.unlock();
		m_resp->status_code = MDS_ENOTOPEN;
		m_resp->msg_type = MDS_FILE_READ_REPLY;
		m_resp->data_len = 0;
		return;
	}
	file_size = cinodes[file_id].inode.size;
	file_bsize = cinodes[file_id].inode.bsize;
	string pathstr(cinodes[file_id].path);
	cinodes_lock.unlock();

	if ((offset >= file_size) || (size <= 0)) {
		m_resp->status_code = MDS_EOVERFLOW;
		m_resp->msg_type = MDS_FILE_READ_REPLY;
		m_resp->data_len = 0;
		return;
	}
	if (offset < 0) {
		offset = 0;
	}
	if ((offset + size) > file_size) {
		size = file_size - offset;
	}
	start_blk = offset / file_bsize;
	end_blk = (offset + size) / file_bsize;
	if (((offset + size) % file_bsize) == 0)
		end_blk--;

	uint64_t *path_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	locks.ReadLock(path_lock_addr);
	struct FileOpREADReply* read_resp = reinterpret_cast<struct FileOpREADReply*>(m_resp->data);
	struct chunk_id *chunklist = reinterpret_cast<struct chunk_id *>(read_resp->chunks);
	read_resp->count = 0;
	struct Fileblk fblk;
	string blkidstr;
	string filedatastr;
	for (int i = start_blk; i <= end_blk; i++) {
		blkidstr = to_string(file_id) + "_" + to_string(i);
		file_data_db->get(blkidstr, &filedatastr);
		fblk.decode(filedatastr);
		chunklist[read_resp->count++] = fblk.chunkid;
	}
	read_resp->data_len = read_resp->count * sizeof(struct chunk_id);
	m_resp->status_code = MDS_SUCCESS;
	m_resp->msg_type = MDS_FILE_READ_REPLY;
	m_resp->data_len = sizeof(struct FileOpREADReply) + read_resp->data_len;
	locks.ReadUnlock(path_lock_addr);
	return;
}

void MDS::handle_write(struct FileOpWRITE *op, struct dss_msg *m_resp)
{
	debug(SECTION_MDS, 10, "handle_write\n");
	uint64_t file_id = op->file_id;
	int64_t offset = op->offset;
	int64_t size = op->size;
	if (size <= 0) {
		m_resp->status_code = MDS_EOVERFLOW;
		m_resp->msg_type = MDS_FILE_WRITE_REPLY;
		m_resp->data_len = 0;
		return;
	}
	if (offset < 0) {
		offset = 0;
	}
	cinodes_lock.lock();
	if (cinodes.find(file_id) == cinodes.end()) {
		cinodes_lock.unlock();
		m_resp->status_code = MDS_ENOTOPEN;
		m_resp->msg_type = MDS_FILE_WRITE_REPLY;
		m_resp->data_len = 0;
		return;
	}
	uint64_t file_size = cinodes[file_id].inode.size;
	uint64_t file_bsize = cinodes[file_id].inode.bsize;
	string pathstr(cinodes[file_id].path);

	debug(SECTION_MDS, 10, "filesize = %d, bsize = %d, openctr = %d .\n", cinodes[file_id].inode.size, cinodes[file_id].inode.bsize, cinodes[file_id].openctr);
	cinodes_lock.unlock();

	debug(SECTION_MDS, 10, "offset = %d, size = %d .\n", offset, size);
	int64_t start_blk, end_blk, file_end_blk;
	start_blk = offset / file_bsize;
	end_blk = (offset + size) / file_bsize;
	if (((offset + size) % file_bsize) == 0)
		end_blk--;
	file_end_blk = file_size / file_bsize;
	if ((file_size % file_bsize) == 0)
		file_end_blk--;

	uint64_t *path_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	locks.WriteLock(path_lock_addr);
	uint64_t tx = transaction.TransBegin();
	transaction.TransLogContext(tx, NULL, 0);
	if ((offset + size) > file_size) {
		cinodes_lock.lock();
		cinodes[file_id].inode.size = offset + size;
		// update kv
		string filekey = to_string(cinodes[file_id].pdir_id) + "_" + cinodes[file_id].inode.name;
		string inodestr;
		cinodes[file_id].inode.encode(&inodestr);
		cinodes_lock.unlock();
		debug(SECTION_MDS, 10, "filekey = %s, inodestr = %s.\n", filekey.c_str(), inodestr.c_str());
		file_meta_db->set(filekey, &inodestr);
	}
	debug(SECTION_MDS, 10, "start_blk = %d, end_blk = %d, file_end_blk = %d .\n", start_blk, end_blk, file_end_blk);
	struct Fileblk fblk;
	/** alloc new chunk **/
	if (end_blk > file_end_blk) {
		for (int64_t i = file_end_blk+1; i <= end_blk; i++) {
			string blkidstr;
			string filedatastr;
			fblk.blkno = i;
			create_chunk(&(fblk.chunkid));
			debug(SECTION_MDS, 10, "blkno = %d, cs_id = %lld, chunk_id = %lld.\n", fblk.blkno, fblk.chunkid.cs_id, fblk.chunkid.blob_id);
			blkidstr = to_string(file_id) + "_" + to_string(i);
			fblk.encode(&filedatastr);
			debug(SECTION_MDS, 10, "blkidstr = %s, filedatastr = %s.\n", blkidstr.c_str(), filedatastr.c_str());
			file_data_db->set(blkidstr, &filedatastr);
		}
	}
	transaction.TransCommit(tx);

	struct FileOpWRITEReply* write_resp = reinterpret_cast<struct FileOpWRITEReply*>(m_resp->data);
	struct chunk_id *chunklist = reinterpret_cast<struct chunk_id *>(write_resp->chunks);
	write_resp->count = 0;
	for (int i = start_blk; i <= end_blk; i++) {
		string blkidstr;
		string filedatastr;
		blkidstr = to_string(file_id) + "_" + to_string(i);
		file_data_db->get(blkidstr, &filedatastr);
		debug(SECTION_MDS, 10, "blkidstr = %s, filedatastr = %s.\n", blkidstr.c_str(), filedatastr.c_str());
		fblk.decode(filedatastr);
		chunklist[write_resp->count++] = fblk.chunkid;
		debug(SECTION_MDS, 10, "cs_id = %d, chunk_id = %d.\n", chunklist[write_resp->count-1].cs_id, chunklist[write_resp->count-1].blob_id);
	}
	write_resp->data_len = write_resp->count * sizeof(struct chunk_id);
	m_resp->status_code = MDS_SUCCESS;
	m_resp->msg_type = MDS_FILE_WRITE_REPLY;
	m_resp->data_len = sizeof(struct FileOpWRITEReply) + write_resp->data_len;
	locks.WriteUnlock(path_lock_addr);
	return;
}

void MDS::handle_truncate(struct FileOpTRUNCATE *op, struct dss_msg *m_resp)
{

}

void MDS::handle_close(struct FileOpCLOSE *op, struct dss_msg *m_resp)
{
	debug(SECTION_MDS, 10, "handle_close\n");
	uint64_t file_id = op->file_id;

	cinodes_lock.lock();
	if (cinodes.find(file_id) == cinodes.end()) {
		cinodes_lock.unlock();
		m_resp->status_code = MDS_ENOTOPEN;
		m_resp->msg_type = MDS_FILE_CLOSE_REPLY;
		m_resp->data_len = 0;
		return;
	}

	bool file_unlink = cinodes[file_id].unlink;
	uint64_t file_size = cinodes[file_id].inode.size;
	uint64_t file_bsize = cinodes[file_id].inode.bsize;
	string pathstr(cinodes[file_id].path);
	string filekey = to_string(cinodes[file_id].pdir_id) + "_" + cinodes[file_id].inode.name;

	if (--cinodes[file_id].openctr == 0) {
		cinodes.erase(file_id);
		cinodes_lock.unlock();
		if (file_unlink) {
			uint64_t *path_lock_addr;
			path_lock_addr = locks.PathToAddress(pathstr);
			locks.WriteLock(path_lock_addr);
			uint64_t tx = transaction.TransBegin();
			transaction.TransLogContext(tx, NULL, 0);
			/** file_meta_db and file_data_db delete**/
			int endblkno = file_size / file_bsize;
			if (file_size % file_bsize == 0)
				endblkno--;
			for (int i = 0; i <= endblkno; i++) {
				string fileblkkey = to_string(file_id) + "_" + to_string(i);
				string fblkstr;
				struct Fileblk fblk;
				file_data_db->get(fileblkkey, &fblkstr);
				fblk.decode(fblkstr);
				delete_chunk(&(fblk.chunkid));
				file_data_db->remove(fileblkkey);
			}
			file_meta_db->remove(filekey);
			transaction.TransCommit(tx);
			locks.WriteUnlock(path_lock_addr);
		}
	} else {
		cinodes_lock.unlock();
	}
	m_resp->status_code = MDS_SUCCESS;
	m_resp->msg_type = MDS_FILE_CLOSE_REPLY;
	m_resp->data_len = 0;
	return;
}

void MDS::handle_fsync(struct FileOpFSYNC *op, struct dss_msg *m_resp)
{

}

void MDS::handle_readdir(struct FileOpREADDIR *op, struct dss_msg *m_resp)
{
	//------ prepare -----------
	char path[MAX_PATH_LENGTH], ppath[MAX_PATH_LENGTH], name[MAX_NAME_LENGTH];
	correct(op->path, path);
	get_ppath_name(path, ppath, name);
	string pathstr(path);
	string ppathstr(ppath);
	string dirstr, pdirstr;
	uint64_t *path_lock_addr, *ppath_lock_addr;
	path_lock_addr = locks.PathToAddress(pathstr);
	ppath_lock_addr = locks.PathToAddress(ppathstr);
	if (0 == strcmp(path, "/")) {
		locks.ReadLock(path_lock_addr);
		dir_meta_db->get(pathstr, &dirstr);
		struct Dir dir;
		dir.decode(dirstr);
		string idstr = to_string(dir.id);
		string dirdatastr;
		dir_data_db->get(idstr, &dirdatastr);
		istringstream iss(dirdatastr);
		string token;
		struct FileOpREADDIRReply* readdir_resp = reinterpret_cast<struct FileOpREADDIRReply*>(m_resp->data);
		struct Dentry *denlist = reinterpret_cast<struct Dentry *>(readdir_resp->dentrys);
		readdir_resp->count = 0;
		while(getline(iss, token, '\n')) {
			denlist[readdir_resp->count++].decode(token);
		}
		readdir_resp->data_len = readdir_resp->count * sizeof(struct Dentry);
		m_resp->status_code = MDS_SUCCESS;
		m_resp->msg_type = MDS_FILE_READDIR_REPLY;
		m_resp->data_len = sizeof(struct FileOpREADDIRReply) + readdir_resp->data_len;
		locks.ReadUnlock(path_lock_addr);
		return;
	}

	locks.ReadLock(path_lock_addr);
	locks.ReadLock(ppath_lock_addr);
	if ( 0 == dir_meta_db->get(pathstr, &dirstr)) {
		struct Dir dir;
		dir.decode(dirstr);
		string idstr = to_string(dir.id);
		string dirdatastr;
		dir_data_db->get(idstr, &dirdatastr);
		istringstream iss(dirdatastr);
		string token;
		struct FileOpREADDIRReply* readdir_resp = reinterpret_cast<struct FileOpREADDIRReply*>(m_resp->data);
		struct Dentry *denlist = reinterpret_cast<struct Dentry *>(readdir_resp->dentrys);
		readdir_resp->count = 0;
		while(getline(iss, token, '\n')) {
			denlist[readdir_resp->count++].decode(token);
		}
		readdir_resp->data_len = readdir_resp->count * sizeof(struct Dentry);
		m_resp->status_code = MDS_SUCCESS;
		m_resp->msg_type = MDS_FILE_READDIR_REPLY;
		m_resp->data_len = sizeof(struct FileOpREADDIRReply) + readdir_resp->data_len;
		locks.ReadUnlock(ppath_lock_addr);
		locks.ReadUnlock(path_lock_addr);
		return;
	} else if (1 == dir_meta_db->get(ppathstr, &pdirstr)) {
		m_resp->status_code = MDS_ENOENT;
		m_resp->msg_type = MDS_FILE_READDIR_REPLY;
		m_resp->data_len = 0;
		locks.ReadUnlock(ppath_lock_addr);
		locks.ReadUnlock(path_lock_addr);
		return;
	}
	struct Dir pdir;
	string pdirdatastr;
	pdir.decode(pdirstr);
	string pidstr = to_string(pdir.id);
	dir_data_db->get(pidstr, &pdirdatastr);
	istringstream iss(pdirdatastr);
	string token;
	while(getline(iss, token, '\n')) {
		struct Dentry d;
		d.decode(token);
		if (strcmp(name, d.name) == 0) {
			m_resp->status_code = MDS_EISFILE;
			m_resp->msg_type = MDS_FILE_READDIR_REPLY;
			m_resp->data_len = 0;
			locks.ReadUnlock(ppath_lock_addr);
			locks.ReadUnlock(path_lock_addr);
			return;
		}
	}
	m_resp->status_code = MDS_ENOENT;
	m_resp->msg_type = MDS_FILE_READDIR_REPLY;
	m_resp->data_len = 0;
	locks.ReadUnlock(ppath_lock_addr);
	locks.ReadUnlock(path_lock_addr);
	return;
}

void MDS::handle_statfs(struct FileOpSTATFS *op, struct dss_msg *m_resp)
{

}

void MDS::handle_rename(struct FileOpRENAME *op, struct dss_msg *m_resp)
{

}

int MDS::create_chunk(struct chunk_id *chunkid)
{
	debug(SECTION_MDS, 10, "---> create_chunk\n");
	struct rpc_msg_info msg_info;
	msg_info.rpc_type = M_CHUNKOP_SERVICE;
	msg_info.client_id = -1;
	msg_info.msg_type = CHUNK_OP_CREATE;
	msg_info.msg_id = 0;
	msg_info.data_len = 0;
	msg_info.data = NULL;
	msg_info.uri = master_uri;
	msg_info.cb = chunk_create_cb;
	msg_info.retval = 0;
	msg_info.ret = (void *) chunkid;
	msg_info.done = false;

	rpc_client_tx_q_lock.lock();
	rpc_client_tx_q.push_back(&msg_info);
	rpc_client_tx_q_lock.unlock();
	{
		unique_lock<mutex> lck(msg_info.lock);
		while (!msg_info.done) {
			msg_info.cond.wait(lck);
		}
	}
	return msg_info.retval;
}

void MDS::chunk_create_cb(void *arg1, void *arg2, void *ctx)
{
	RdmaRpc *rpc = static_cast<RdmaRpc *>(arg1);
	RpcExecution *ex = static_cast<RpcExecution *>(arg2);
	struct rpc_msg_info *msg_info = static_cast<struct rpc_msg_info *>(ctx);
	RpcMsg *rpc_resp = ex->resp_msg;
	struct dss_msg *dss_resp = (struct dss_msg *)rpc_resp->data;

	if (dss_resp->status_code == SUCCESS) {
		debug(SECTION_MDS, 10, "alloc chunk success\n");
		struct chunk_id *chunkidret = reinterpret_cast<struct chunk_id *>(dss_resp->data);
		struct chunk_id *chunkid = reinterpret_cast<struct chunk_id *>(msg_info->ret);
		chunkid->cs_id = chunkidret->cs_id;
		chunkid->blob_id = chunkidret->blob_id;
	} else {
		debug(SECTION_MDS, 10, "alloc chunk from chunkserver faild, errno = %d \n", dss_resp->status_code);
		msg_info->retval = -1;
	}
	{
		unique_lock<mutex> lck(msg_info->lock);
		msg_info->done = true;
		msg_info->cond.notify_one();
	}
}

int MDS::delete_chunk(struct chunk_id *chunkid)
{
	debug(SECTION_MDS, 10, "---> delete_chunk\n");
	struct rpc_msg_info msg_info;
	msg_info.rpc_type = S_CHUNK_IO_SERVICE;
	msg_info.client_id = -1;
	msg_info.msg_type = CHUNK_OP_DELETE;
	msg_info.msg_id = 0;
	msg_info.data_len = sizeof(struct chunk_req);
	struct chunk_req request;
	request.id.cs_id = chunkid->cs_id;
	request.id.blob_id = chunkid->blob_id;
	msg_info.data = &request;
	msg_info.uri = cs_uri[chunkid->cs_id];
	msg_info.cb = chunk_delete_cb;
	msg_info.retval = 0;
	msg_info.ret = NULL;
	msg_info.done = false;

	rpc_client_tx_q_lock.lock();
	rpc_client_tx_q.push_back(&msg_info);
	rpc_client_tx_q_lock.unlock();
	{
		unique_lock<mutex> lck(msg_info.lock);
		while (!msg_info.done) {
			msg_info.cond.wait(lck);
		}
	}
	return msg_info.retval;
}

void MDS::chunk_delete_cb(void *arg1, void *arg2, void *ctx)
{
	RdmaRpc *rpc = static_cast<RdmaRpc *>(arg1);
	RpcExecution *ex = static_cast<RpcExecution *>(arg2);
	struct rpc_msg_info *msg_info = static_cast<struct rpc_msg_info *>(ctx);
	RpcMsg *rpc_resp = ex->resp_msg;
	struct dss_msg *dss_resp = (struct dss_msg *)rpc_resp->data;
	
	if (dss_resp->status_code == SUCCESS) {
		msg_info->retval = 0;
	} else {
		msg_info->retval = -1;
	}
	{
		unique_lock<mutex> lck(msg_info->lock);
		msg_info->done = true;
		msg_info->cond.notify_one();
	}
}


